package com.edu.realtime.util;

import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.common.EduConfig;

import redis.clients.jedis.Jedis;

/**
 * @author zcx
 * @create 2022-10-18 22:20 维度查询的工具类
 */
public class DimUtil {
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName, Tuple2.of("id", id));
    }

    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        // 拼接从Redis中查询维度的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        // 拼接查询维度的语句
        StringBuilder selectSql =
            new StringBuilder("select * from " + EduConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        // 对所有的查询进行遍历
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName).append("='").append(columnValue).append("' ");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        // 先从Redis中获取维度数据
        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            dimInfoStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("~~从Redis中查询维度数据发生了异常~~");
        }

        // 判断是否缓存命中
        if (StringUtils.isNotEmpty(dimInfoStr)) {
            // 缓存命中，直接将字符串转换为json对象，作为查询的维度返回
            dimJsonObj = JSON.parseObject(dimInfoStr);
        } else {
            // 缓存未命中，发送请求到phoenix表中查询维度数据，并将查询的维度放到Redis中缓存起来
            System.out.println("从phoenix表中查询维度数据的SQL：" + selectSql);
            // 底层还是调用的PhoenixUtil工具类中的queryList方法
            List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            // 因为我们是根据维度的主键去查询维度数据，所以如果查到维度数据的话，集合中也只会有一条数据
            if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
                dimJsonObj = dimInfoJsonObjList.get(0);
                // 将查询的维度放到Redis中缓存起来 并设置失效时间
                if (jedis != null) {
                    jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObj.toJSONString());
                }
            } else {
                System.out.println("在维度表中没有找到对应的维度数据");
            }
        }

        // 释放资源
        if (jedis != null) {
            System.out.println("~~关闭Jedis客户端~~");
            jedis.close();
        }
        return dimJsonObj;
    }

    // 根据维度的主键 到维度表中查询维度数据 没有任何优化
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        // 拼接查询维度的语句
        StringBuilder selectSql =
            new StringBuilder("select * from " + EduConfig.PHOENIX_SCHEMA + "." + tableName + " where ");
        // 对所有的查询进行遍历
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "' ");
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
            }
        }
        System.out.println("从phoenix表中查询维度数据的SQL:" + selectSql);
        // 底层还是调用的PhoenixUtil工具类中的queryList方法
        List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        // 因为我们是根据维度的主键去查询维度数据，所以如果查到维度数据的话，集合中也只会有一条数据
        JSONObject dimJsonObj = null;
        if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
            dimJsonObj = dimInfoJsonObjList.get(0);
        } else {
            System.out.println("在维度表中没有找到对应的维度数据");
        }
        return dimJsonObj;
    }

    public static void deleteCached(String tableName, String id) {
        String redisKey = "dim:" + tableName.toLowerCase() + ":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                System.out.println("~~删除缓存后，关闭Jedis连接~~");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) {
        System.out.println(getDimInfo("dim_course_info", "1"));
    }
}
